package com.weyoung.app.consumer;

import com.rabbitmq.client.Channel;
import com.weyoung.app.dto.req.CultureViewUpdateDTO;
import com.weyoung.app.dto.req.FamilyCultureMessage;
import com.weyoung.app.mapper.FamilyCultureMapper;
import com.weyoung.common.config.RabbitMQConfig;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component
@Slf4j
@RequiredArgsConstructor
public class FamilyCultureConsumer {

    // 参数控制
    private static final int MAX_BATCH_SIZE = 100; // 最大批量大小
    private static final long FLUSH_INTERVAL_S = 60; // 60秒刷新间隔

    // 依赖注入
    private final StringRedisTemplate redisTemplate;
    private final FamilyCultureMapper cultureMapper;
//    private final

    // 批量处理相关配置
    // 批处理大小
    private static final int BATCH_SIZE = 10;
    // 缓存列表
    private final List<FamilyCultureMessage> batchBuffer = Collections.synchronizedList(new ArrayList<>());
    // 线程池
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    @PostConstruct
    public void init() {
        log.info("FamilyCultureConsumer 初始化完成，开始调度定时任务");
        // 定时刷新缓冲区
        scheduler.scheduleAtFixedRate(this::flushBatch, 5, FLUSH_INTERVAL_S, TimeUnit.SECONDS);
    }

    @PreDestroy
    public void destroy() {
        scheduler.shutdown();
        // 应用关闭前强制刷新剩余消息
        flushBatch();
    }

    @RabbitListener(queues = RabbitMQConfig.FAMILY_CULTURE_QUEUE_NAME)
    public void handleMessage(FamilyCultureMessage message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        log.info("【用户浏览量队列】收到消息：{}", message);
        try {

            // 2. 原子性累加
            synchronized (batchBuffer){
                batchBuffer.add(message);
                if (batchBuffer.size() >= BATCH_SIZE){
                    flushBatch();
                }
            }

            // 3. 触发批量条件
            if (batchBuffer.size() >= MAX_BATCH_SIZE) {
                flushBatch();
            }

            // 4. 确认消息
            channel.basicAck(tag, false);

        } catch (Exception e) {
            log.error("处理消息失败: {}", message.getCultureId(), e);
            try {
                channel.basicNack(tag, false, true); // 重试
            } catch (IOException ex) {
                log.error("NACK失败", ex);
            }
        }
    }

    private void flushBatch() {

        List<FamilyCultureMessage> messages;

        // 1. 获取缓冲区快照并清空
        synchronized (batchBuffer) {
            if (batchBuffer.isEmpty()) {
                log.info("【用户浏览量队列】批量刷新触发，缓冲区为空，跳过处理");
                return;
            }
            messages = new ArrayList<>(batchBuffer);
            batchBuffer.clear();
        }

        // 2. 按cultureId分组统计浏览量
        Map<String, Long> viewCountMap = messages.stream()
                .collect(Collectors.groupingBy(
                        FamilyCultureMessage::getCultureId,
                        Collectors.counting()
                ));

        if (viewCountMap.isEmpty()) {
            return;
        }

        log.info("【用户浏览量队列】批量处理 {} 条消息，涉及 {} 个文化活动",
                messages.size(), viewCountMap.size());

        // 3. 分批更新数据库
        try {
            // 3.1 将Map转换为批量更新DTO列表
            List<CultureViewUpdateDTO> updateList = viewCountMap.entrySet().stream()
                    .map(entry -> new CultureViewUpdateDTO(entry.getKey(), entry.getValue()))
                    .collect(Collectors.toList());

            // 3.2 分批执行更新（每批10条）
            int batchSize = 10;
            for (int i = 0; i < updateList.size(); i += batchSize) {
                List<CultureViewUpdateDTO> batch = updateList.subList(
                        i, Math.min(i + batchSize, updateList.size()));

                // 3.3 执行批量更新
                int affectedRows = cultureMapper.batchIncrementViewCount(batch);

                if (affectedRows < batch.size()) {
                    log.warn("部分更新未生效，预期更新 {} 条，实际更新 {} 条", batch.size(), affectedRows);
                    // 可以将失败记录重新放回缓冲区或记录日志
                }
            }

            log.info("【用户浏览量队列】批量处理完成，共更新 {} 个文化活动的浏览量", updateList.size());
        } catch (Exception e) {
            log.error("【用户浏览量队列】批量更新数据库失败", e);

            // 4. 失败处理：将未处理的消息重新放回缓冲区
            synchronized (batchBuffer) {
                batchBuffer.addAll(messages);
                log.warn("已将 {} 条消息重新放回缓冲区", messages.size());
            }
        }
    }

}
